use std::io::BufRead;
use std::sync::{Arc, mpsc};
use anyhow::Error;
use dashmap::{DashMap, DashSet};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, ReadBuf};
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use common::{Msg, MsgType};


const LOCAL_SERVER: &str = "127.0.0.1:8088";

//
#[tokio::main]
async fn main() -> anyhow::Result<(), Error> {


    let tcp_listener = TcpListener::bind(LOCAL_SERVER).await?;


    let (mut tx, rx) = broadcast::channel(16);


    let mut user_hash = Arc::new(DashMap::new());
    //
    //
    let mut user_set = Arc::new(DashSet::new());


    loop {

        // accept 是监听所有的连接, 所以要放在loop里面
        let (mut socket, addr) = tcp_listener.accept().await?;

        let mut user_hash_clone = Arc::clone(&user_hash);
        user_hash_clone.insert(addr.to_string(), "".to_string());
        //
        let user_set_clone = Arc::clone(&user_set);

        let tx = tx.clone();

        let mut rx = tx.subscribe();

        // println!("{} is up ", addr.to_string());

        tokio::spawn(async move {
            let (reader, mut writer) = socket.split();

            let mut reader = BufReader::new(reader);


            let mut msg = String::new();

            loop {
                tokio::select! {
                    result = reader.read_line(&mut msg) => {
                        if  result.unwrap() ==0 {

                            break;
                        }
                        // println!("read  tcp stream msg is {} , addr is {:?}", msg, addr);


                        let msg_obj: Msg  = Msg::parse(msg.as_str());

                        match msg_obj.msg_type {

                            MsgType::SIGN_UP => {

                                // println!("msg_obj is {:?}", msg_obj);

                                // println!("user_hash is {:?},  user_set is {:?}" , user_hash_clone , user_set_clone) ;


                                if user_set_clone.contains(&msg_obj.data.clone()) {

                                    let  msg_obj = Msg::sign_up_fail("昵称重复了,请重新输入");

                                    writer.write_all(msg_obj.to_string().as_bytes()).await.unwrap();

                                }else{
                                    let nick_name  = msg_obj.data.replace("\r","") ;
                                    user_hash_clone.insert(addr.to_string() , nick_name.clone());

                                    user_set_clone.insert(msg_obj.data.clone());

                                    let  msg_obj = Msg::sign_up_success();

                                    writer.write_all(msg_obj.to_string().as_bytes()).await.unwrap();


                                    let  msg_obj = Msg::sign_up_broadcast(nick_name.clone());

                                    tx.send((msg_obj.to_string(), addr));
                                }
                            }

                            MsgType::NOT_SUPPORT => {

                                println!("msg type is NOT_SUPPORT");
                            }


                            // MsgType::QUIT => {
                            //
                            //     user_hash_clone.remove(&addr.to_string());
                            //
                            //     user_set_clone.remove(&msg_obj.data.clone());
                            //
                            //     tx.send((msg.clone(), addr));
                            // }

                            _=> {
                                tx.send((msg.clone(), addr));
                            }
                        }
                        msg.clear()

                    }

                    result = rx.recv() => {

                        let (msg , other_addr ) =  result.unwrap() ;

                        println!("receive msg  from channel , msg is {} , addr is {} , my addr is {}", msg , other_addr, addr );


                        let mut msg_obj: Msg  = Msg::parse(msg.as_str());


                            if addr == other_addr {
                                continue;
                            }

                        let  nick_name  = user_hash_clone.get(&other_addr.to_string()).unwrap().value().to_string();
                        msg_obj.nick_name =  nick_name;

                        println!(" msg_obj is {:?}", msg_obj);

                        writer.write_all(msg_obj.to_string().as_bytes()).await.unwrap();
                    }

                }

            }


            // tcp 连接断开
            println!("addr {}  is disconnect+++++++++++", addr);

            //
            // let nick_name = user_hash_clone.get(&addr.to_string()).unwrap();
            // let msg_quit = Msg::quit2((*nick_name.value()).clone());
            //
            // user_hash_clone.remove(&addr.to_string());
            //
            // user_set_clone.remove(nick_name.value());
            //
            // tx.send((msg_quit.to_string(), addr));
        });

    }


    Ok(())
}
